package Coordinador;

import Tipos.*;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Encargado de actuar como coordinador entre los participantes
 * de una transaccion distribuida en este caso coordina el commit de 2 fases
 * entre las panaderias.
 */
public class Coordinador {    

    private static final int CANTPANADERIAS = 5;
    
    public static void main(String[] args) {
        try {
            int puertoCoordinador = 4020;
            ServerSocket serverSocket = new ServerSocket(puertoCoordinador);   
            
            List<Integer> listaCommit = new LinkedList<>();// Lista que almacena el idCommit de los participantes que votaron por hacer commit y estan esperando que llegue el voto del otro participante
            List<Integer> listaMensajesEsperadosIgnorar = new LinkedList<>();//Lista que almacena el idCommit de los mensajes que deben llegar y que debo ignorar, pues su participante voto por abortar la transaccion
            List<RespuestaEsperada> listaRespuestasEsperadas = new LinkedList<>();//Lista que almacena los mensajes de respuesta que deben mandar los participantes del commit con el tiempo en que se mandan los mensajes de vote_request. 
            HashMap<Integer,UbicacionPanaderia> panaderiasAbiertas = new HashMap<Integer,UbicacionPanaderia>();//Lista que almacena la direccione de las panaderias abiertas
            HashMap<Integer,ObjectOutputStream> listaOutputStreamPanaderias = new HashMap<Integer,ObjectOutputStream>();//Lista que almacena los streams de salida de cada panaderia
            Buffer bufferPeticiones = new Buffer();//buffer para que el escuchador encole los mensajes
            UbicacionPanaderia ubicacionCoordinador = new UbicacionPanaderia(serverSocket.getInetAddress().getHostAddress(), puertoCoordinador, "Coordinador", 0);// Ubicacion del coordinador
            int globalIdCommit = 1;//Variable que almacena el ultimo idCommit. Sirve para darle un id unico a cada operacion de commit para poder identificarlas mas facilmente
            boolean cierre = false;//Sirve para saber cuando debe cerrar el coordinador, se pone en true cuando la lista panaderiasAbiertas es vacia
            
            for (int i = 1; i <= CANTPANADERIAS; i++) {
                Socket cliente ;
                cliente = serverSocket.accept();
                UbicacionPanaderia ubicacion = new UbicacionPanaderia(cliente.getInetAddress().getHostAddress(), cliente.getPort(), "Panaderia_" + i, i);
                
                System.out.println("Coordinador---> se conecto la panaderia con direccion: "+ ubicacion.toString());
                //System.out.println("Coordinador---> se conecto la panaderia con puerto: " + cliente.getPort() + " y puerto local: " + cliente.getLocalPort());
                panaderiasAbiertas.put(ubicacion.getId(), ubicacion);
                ObjectOutputStream outStreamEmisor = new ObjectOutputStream(cliente.getOutputStream());
                ObjectInputStream inStream = new ObjectInputStream(cliente.getInputStream());
                listaOutputStreamPanaderias.put(i, outStreamEmisor);
                CoordinadorEscuchadorTCP escuchadorTCP = new CoordinadorEscuchadorTCP("Coordinador_Escuchador_Panaderia_" + i, bufferPeticiones, inStream); 
                escuchadorTCP.start();
            }
            
            for (int j = 1; j<= CANTPANADERIAS; j++) {
                ObjectOutputStream outStreamEmisor = listaOutputStreamPanaderias.get(j);
                Mensaje mensaje = new Mensaje();
                mensaje.setMensajePanaderia(TipoMensaje.COMENZAR_PANADERIA);
                mensaje.setCantPanes(0);
                mensaje.setParticipanteCommit(panaderiasAbiertas.get(j));
                mensaje.setEmisor(ubicacionCoordinador);
                mensaje.setIniciador(ubicacionCoordinador);
                outStreamEmisor.writeObject(mensaje);
                System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensaje.mostrarPantalla());
            }
            
            
            //System.out.println("despues de agregar a todas las panaderias---->listaUbicacion.size: " + listaUbicacionPanaderia.size());
            while(!cierre){
                controlarTiempos(System.currentTimeMillis(), listaRespuestasEsperadas, listaMensajesEsperadosIgnorar, listaOutputStreamPanaderias, ubicacionCoordinador);
                if(!bufferPeticiones.isEmpty()){
                    Mensaje mensaje = bufferPeticiones.get();
                    ObjectOutputStream outStreamEmisor = listaOutputStreamPanaderias.get(mensaje.getEmisor().getId());
                    System.out.println("CoordinadorProcesador---> procesando mensaje: "+ mensaje.mostrarPantalla());
                    ObjectOutputStream outStreamParticipanteCommit = null;
                    if (mensaje.getParticipanteCommit()!=null)
                        outStreamParticipanteCommit = listaOutputStreamPanaderias.get(mensaje.getParticipanteCommit().getId());
                    boolean ignorar = false;
                    if(mensaje.getMensajePanaderia() == TipoMensaje.VOTE_ABORT || mensaje.getMensajePanaderia() == TipoMensaje.VOTE_COMMIT){
                        if(listaMensajesEsperadosIgnorar.contains(mensaje.getIdCommit())){
                            ignorar = true;
                            boolean borrado = listaMensajesEsperadosIgnorar.remove(new Integer(mensaje.getIdCommit()));
                            System.out.println("Eliminado idCommit = " + mensaje.getIdCommit() + " " + borrado);
                            System.out.println("MensajeIgnorado: " + mensaje.mostrarPantalla() + " Ignorar = " + ignorar);
                            RespuestaEsperada respuesta = new RespuestaEsperada(000, mensaje.getIdCommit(), mensaje.getEmisor(), mensaje.getParticipanteCommit(), mensaje.getIniciador());
                            listaRespuestasEsperadas.remove(respuesta);
                        }
                    }
                    if(!ignorar){
                        switch (mensaje.getMensajePanaderia()){
                            case DO_COMMIT:{
                                if(!respuestaPendiente(listaRespuestasEsperadas, mensaje, bufferPeticiones)){
                                    Mensaje mensajeParticipante = new Mensaje();
                                    Mensaje mensajeEnviar = new Mensaje();
                                    mensajeParticipante.setIdCommit(globalIdCommit);
                                    mensajeParticipante.setMensajePanaderia(TipoMensaje.VOTE_REQUEST);
                                    mensajeParticipante.setCantPanes(mensaje.getCantPanes());
                                    mensajeParticipante.setEmisor(ubicacionCoordinador);
                                    mensajeParticipante.setIniciador(mensaje.getIniciador());
                                    mensajeParticipante.setParticipanteCommit(mensaje.getEmisor());
                                    if(panaderiasAbiertas.get(mensaje.getParticipanteCommit().getId()) != null){
                                        outStreamParticipanteCommit.writeObject(mensajeParticipante);
                                        System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeParticipante.mostrarPantalla());
                                        mensajeEnviar.setMensajePanaderia(TipoMensaje.VOTE_REQUEST);
                                        RespuestaEsperada respuestaEmisor = new RespuestaEsperada(System.currentTimeMillis(), globalIdCommit, mensaje.getEmisor(), mensaje.getParticipanteCommit(), mensaje.getIniciador());
                                        RespuestaEsperada respuestaPCommit = new RespuestaEsperada(System.currentTimeMillis(), globalIdCommit, mensaje.getParticipanteCommit(), mensaje.getEmisor(), mensaje.getIniciador());
                                        listaRespuestasEsperadas.add(respuestaEmisor);
                                        listaRespuestasEsperadas.add(respuestaPCommit);
                                    }else{
                                        mensajeEnviar.setMensajePanaderia(TipoMensaje.GLOBAL_ABORT);
                                    }
                                    mensajeEnviar.setIdCommit(globalIdCommit);
                                    mensajeEnviar.setCantPanes(mensaje.getCantPanes());
                                    mensajeEnviar.setEmisor(ubicacionCoordinador);
                                    mensajeEnviar.setIniciador(mensaje.getIniciador());
                                    mensajeEnviar.setParticipanteCommit(mensaje.getParticipanteCommit());
                                    outStreamEmisor.writeObject(mensajeEnviar);
                                    System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeEnviar.mostrarPantalla());
                                    globalIdCommit++;
                                }
                                break;
                            }
                            case VOTE_COMMIT:{
                                if (!listaCommit.contains(mensaje.getIdCommit())){
                                    listaCommit.add(mensaje.getIdCommit());
                                    RespuestaEsperada respuesta = new RespuestaEsperada(00, mensaje.getIdCommit(), mensaje.getEmisor(), mensaje.getParticipanteCommit(), mensaje.getIniciador());
                                    System.out.println("Respuesta eliminada? " + listaRespuestasEsperadas.remove(respuesta));
                                }else{
                                    RespuestaEsperada respuesta = new RespuestaEsperada(00, mensaje.getIdCommit(), mensaje.getEmisor(), mensaje.getParticipanteCommit(), mensaje.getIniciador());
                                    System.out.println("Respuesta eliminada? " + listaRespuestasEsperadas.remove(respuesta));
                                    System.out.println("por enviar GLOBAL_COMMIT");
                                    Mensaje mensajeEnviar = new Mensaje();
                                    mensajeEnviar.setIdCommit(mensaje.getIdCommit());
                                    mensajeEnviar.setMensajePanaderia(TipoMensaje.GLOBAL_COMMIT);
                                    mensajeEnviar.setCantPanes(mensaje.getCantPanes());
                                    mensajeEnviar.setEmisor(ubicacionCoordinador);
                                    mensajeEnviar.setIniciador(mensaje.getIniciador());
                                    mensajeEnviar.setParticipanteCommit(mensaje.getParticipanteCommit());
                                    outStreamEmisor.writeObject(mensajeEnviar);
                                    System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeEnviar.mostrarPantalla());
                                    Mensaje mensajeParticipante = new Mensaje();
                                    mensajeParticipante.setIdCommit(mensaje.getIdCommit());
                                    mensajeParticipante.setMensajePanaderia(TipoMensaje.GLOBAL_COMMIT);
                                    mensajeParticipante.setCantPanes(mensaje.getCantPanes());
                                    mensajeParticipante.setEmisor(ubicacionCoordinador);
                                    mensajeParticipante.setIniciador(mensaje.getIniciador());
                                    mensajeParticipante.setParticipanteCommit(mensaje.getEmisor());
                                    outStreamParticipanteCommit.writeObject(mensajeParticipante);
                                    System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeParticipante.mostrarPantalla());
                                    listaCommit.remove(new Integer(mensaje.getIdCommit()));
                                }  
                                break;
                            }
                            case VOTE_ABORT:{
                                RespuestaEsperada respuesta = new RespuestaEsperada(00, mensaje.getIdCommit(), mensaje.getEmisor(), mensaje.getParticipanteCommit(), mensaje.getIniciador());
                                System.out.println("Respuesta eliminada? " + listaRespuestasEsperadas.remove(respuesta));
                                boolean esperarMensaje = false;
                                System.out.println("por enviar GLOBAL_ABORT");
                                Mensaje mensajeParticipante = new Mensaje();
                                mensajeParticipante.setIdCommit(mensaje.getIdCommit());
                                mensajeParticipante.setMensajePanaderia(TipoMensaje.GLOBAL_ABORT);
                                mensajeParticipante.setCantPanes(mensaje.getCantPanes());
                                mensajeParticipante.setEmisor(ubicacionCoordinador);
                                mensajeParticipante.setIniciador(mensaje.getIniciador());
                                mensajeParticipante.setParticipanteCommit(mensaje.getEmisor());
                                if(panaderiasAbiertas.get(mensaje.getParticipanteCommit().getId()) != null){
                                    outStreamParticipanteCommit.writeObject(mensajeParticipante);
                                    System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeParticipante.mostrarPantalla());
                                    esperarMensaje = true;//si el participante commit esta cerrado significa que no debo esperar ningun mensaje de esa sucursal
                                }    
                                Mensaje mensajeEnviar = new Mensaje();
                                mensajeEnviar.setIdCommit(mensaje.getIdCommit());
                                mensajeEnviar.setMensajePanaderia(TipoMensaje.GLOBAL_ABORT);
                                mensajeEnviar.setCantPanes(mensaje.getCantPanes());
                                mensajeEnviar.setEmisor(ubicacionCoordinador);
                                mensajeEnviar.setIniciador(mensaje.getIniciador());
                                mensajeEnviar.setParticipanteCommit(mensaje.getParticipanteCommit());
                                outStreamEmisor.writeObject(mensajeEnviar);
                                System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeEnviar.mostrarPantalla());
                                if(!listaCommit.remove(new Integer(mensaje.getIdCommit())) && esperarMensaje){
                                    listaMensajesEsperadosIgnorar.add(mensaje.getIdCommit());
                                    System.out.println("Ignorar mensaje: " + mensaje.getIdCommit());
                                }else if(!listaCommit.remove(new Integer(mensaje.getIdCommit())) && !esperarMensaje){//nunca me va a llegar el mensaje de participante commit porque ya cerro
                                            RespuestaEsperada respuestaEsperada = new RespuestaEsperada(00, mensaje.getIdCommit(), mensaje.getParticipanteCommit(), mensaje.getEmisor(), mensaje.getIniciador());
                                            System.out.println("Respuesta eliminada? " + listaRespuestasEsperadas.remove(respuestaEsperada));
                                        }
                                break;
                            }
                            case PANADERIA_CERRADA:{
                                //System.out.println("listaUbicacion.size Antes= " + listaUbicacionPanaderia.size());
                                Mensaje mensajeEnviar = new Mensaje();
                                mensajeEnviar.setMensajePanaderia(TipoMensaje.PANADERIA_CERRADA);
                                mensajeEnviar.setCantPanes(mensaje.getCantPanes());
                                mensajeEnviar.setEmisor(ubicacionCoordinador);
                                mensajeEnviar.setIniciador(mensaje.getIniciador());
                                mensajeEnviar.setParticipanteCommit(mensaje.getParticipanteCommit());
                                outStreamEmisor.writeObject(mensajeEnviar);
                                System.out.println("CoordinadorProcesador---> Envie mensaje: "+ mensajeEnviar.mostrarPantalla());
                                panaderiasAbiertas.remove(mensaje.getEmisor().getId());
                                listaOutputStreamPanaderias.remove(mensaje.getEmisor().getId());
                                //System.out.println("listaUbicacion.size Despues= " + listaUbicacionPanaderia.size());
                                if (panaderiasAbiertas.isEmpty())
                                    cierre = true;
                                break;                
                            }                  
                       }
                    }
                                        
                }
            }
            System.out.println("listaRespuestasEsperadas.size = " + listaRespuestasEsperadas.size());
            System.out.println("Cerro el Coordinador");
            System.exit(0);
        } catch (IOException ex) {
            Logger.getLogger(Coordinador.class.getName()).log(Level.SEVERE, null, ex);
        }
     }

    private static void controlarTiempos(long currentTimeMillis, List<RespuestaEsperada> listaRespuestasEsperadas, List<Integer> listaMensajesEsperadosIgnorar, HashMap<Integer,ObjectOutputStream> listaOutputStreamPanaderias, UbicacionPanaderia ubicacionCoordinador) {
        List<RespuestaEsperada> listaRespuestasBorrar = new LinkedList<>();
        for (Iterator<RespuestaEsperada> it = listaRespuestasEsperadas.iterator(); it.hasNext();) {
            RespuestaEsperada respuesta = it.next();
            long diferenciaTiempo = currentTimeMillis - respuesta.getTiempoIngreso();
            if(diferenciaTiempo >= 5000){
                listaMensajesEsperadosIgnorar.add(respuesta.getIdCommit());
                Mensaje mensajeEnviar = new Mensaje();
                mensajeEnviar.setIdCommit(respuesta.getIdCommit());
                mensajeEnviar.setEmisor(ubicacionCoordinador);
                mensajeEnviar.setIniciador(respuesta.getIniciador());
                mensajeEnviar.setParticipanteCommit(respuesta.getEmisor());
                mensajeEnviar.setMensajePanaderia(TipoMensaje.GLOBAL_ABORT);
                ObjectOutputStream outStreamParticipante = listaOutputStreamPanaderias.get(respuesta.getParticipanteCommit().getId());
                System.out.println("Expiro el tiempo: " + mensajeEnviar.mostrarPantalla());
                try {
                    if(outStreamParticipante != null)
                        outStreamParticipante.writeObject(mensajeEnviar);// se la mando al participanteCommit, porque si nunca vino el mnensaje con el emisor quiere decir que el emisor puede haberse cerrado
                    else
                        System.out.println("La panaderia ya cerro el socket");
                } catch (IOException ex) {
                    Logger.getLogger(Coordinador.class.getName()).log(Level.SEVERE, mensajeEnviar.mostrarPantalla(), ex);
                }
                listaRespuestasBorrar.add(respuesta);
            }
        }
        for (Iterator<RespuestaEsperada> it = listaRespuestasBorrar.iterator(); it.hasNext();) {
            RespuestaEsperada respuestaBorrar = it.next();
            listaRespuestasEsperadas.remove(respuestaBorrar);
        }
    }

    private static boolean respuestaPendiente(List<RespuestaEsperada> listaRespuestasEsperadas, Mensaje mensaje, Buffer bufferPeticiones) {
        for (Iterator<RespuestaEsperada> it = listaRespuestasEsperadas.iterator(); it.hasNext();) {
            RespuestaEsperada respuesta = it.next();
            if(respuesta.getEmisor().equals(mensaje.getEmisor()) || (respuesta.getEmisor().equals(mensaje.getParticipanteCommit())) || respuesta.getParticipanteCommit().equals(mensaje.getEmisor()) || (respuesta.getParticipanteCommit().equals(mensaje.getParticipanteCommit()))){
                bufferPeticiones.put(mensaje);
                return true;
            }
        }
        return false;
    }
}
